2025.5.26 RNNコンソール版【torch】

ライプラリのインポート

code:rnn.py

import pandas as pd

import matplotlib.pyplot as plt

import numpy as np

# sklearnから

import sklearn.preprocessing as slpr

import sklearn.model_selection as slms

# PyTorchから

import torch as pt

from torch import nn, optim

from torch.utils.data import DataLoader, TensorDataset, Dataset

# from torchvision import transforms

from torchinfo import summary

import torchinfo

# 画像

from PIL import Image

import requests, io, os

# 結果の保存用ディレクトリ

dirname = './RNN/'

os.makedirs(dirname, exist_ok=True)

データセットの取得

Kaggleの提供するpassengersデータセットを利用する。周期性が高い。

正規化し、値の範囲が0~1に納まるようにする。

code:続き.py

# データ取得

trans_normal = slpr.MinMaxScaler(feature_range=(0, 1)) # 正規化器

data_normal = trans_normal.fit_transform(data.reshape(-1, 1)) # 正規化器を使って正規化

plt.figure(figsize=(10, 3), dpi=100)

plt.grid()

plt.plot(data_normal)

plt.savefig(dirname + 'train_data.png')

plt.clf()

訓練データの用意

上で用意したデータセットから訓練データを格納したDataloaderオブジェクトを生成する。

１つの訓練データの長さをwindows_sizeで指定する。

ウインドウを元の時系列データの先頭から１ステップづつ進めることで生成する。件数をtrain_nに格納。

訓練に用いる学習・教師データの入れ物に連続代入する。

code:続き.py

data_train, data_test = slms.train_test_split(data_normal, test_size=0.3, shuffle=False)

print('# 訓練データ：', data_train.shape)

print('# 検証データ：', data_test.shape)

window_size = 10

train_n = len(data_train) - window_size - 1

X = np.zeros((train_n, window_size, 1))

y = np.zeros((train_n, 1))

for i in range(train_n):

X = pt.tensor(X, dtype=pt.float)

y = pt.tensor(y, dtype=pt.float)

train_dataset = pt.utils.data.TensorDataset(X, y)

train_loader = DataLoader(train_dataset, batch_size=4, shuffle=True)

RNNの定義

batch_firstを有効にしている。与えるテンソルの構造に注意。

X［バッチ］［データ長］［特徴量］

この場合、

データ長は10

1つの系列のデータなので、特徴量の数は1

code:続き.py

class MyRNN(nn.Module):

def __init__(self, input_size, output_size, hidden_dim, n_layers):

super(MyRNN, self).__init__()

self.input_size = input_size

self.hidden_dim = hidden_dim

self.n_layers = n_layers

self.rnn = nn.RNN(input_size, hidden_dim, n_layers, batch_first=True)

self.fc = nn.Linear(hidden_dim, output_size)

def forward(self, x):

y_rnn, h = self.rnn(x, None)

return y

main部

ここからプログラムの処理が始まる

code:続き.py

# RNNの設定

print('# Result is outout in' + dirname)

n_inputs = 1

n_outputs = 1

n_hidden = 1

n_layers = 1

epochs = 1000

net = MyRNN(n_inputs, n_outputs, n_hidden, n_layers) # ネット

func_loss = nn.MSELoss() # 損失関数

optimizer = optim.Adam(net.parameters(), lr=0.001) # 最適化器

loss_history = []

# 開始

for i in range(epochs + 1):

net.train() # NNを訓練モードにセット

tmp_loss = 0.0

for j, (x, t) in enumerate(train_loader):

optimizer.zero_grad()

y = net(x)

loss = func_loss(y, t)

loss.backward()

optimizer.step()

tmp_loss += loss.item()

tmp_loss /= j + 1

loss_history.append(tmp_loss)

if i%100 == 0:

print('Epoch:', i, 'Loss_Train:', tmp_loss)

predict = []

for k in range(train_n): # k ... 0:89

y = net(x_input.reshape(1, window_size, 1)) # (1, -1, 1)でもよし

predict.append(y.item())

data_step = range(len(data_train)) # 0:100 (100 steps)

predict_step = range(window_size, window_size +　len(predict))　　　# 10:99 (89 steps)

plt.plot(data_step, data_train, label='Correct')

plt.plot(predict_step, predict, label='Predicted')

filename = 'fig_RNN' + str(i).zfill(5) + '.png' # ファイル名

plt.savefig(dirname + filename)

plt.clf()

plt.plot(loss_history)

plt.savefig(dirname + 'loss.png')